package com.mjf.spark.day11

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 无状态转化操作-Transform
 *    使用Transform算子将DS转换为rdd
 */
object SparkStreaming01_Transform {
  def main(args: Array[String]): Unit = {

    // 创建配置文件对象   注意：Streaming程序执行至少需要2个线程（1个用来采集，1一个用来计算），所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")

    // 创建SparkStreaming程序执行入口对象   Seconds:时间间隔
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 从指定的端口获取数据
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop103", 9999)

    // 将DS转换为RDD进行操作    DS本身是没有提供排序函数的
    val resDS: DStream[(String, Int)] = socketDS.transform(
      rdd => {
        val flatMapRDD: RDD[String] = rdd.flatMap(_.split(" ")) // 这里是对DStream底层封装的RDD进行操作
        val mapRDD: RDD[(String, Int)] = flatMapRDD.map((_, 1))
        val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
        reduceRDD.sortByKey()
      }
    )

    /*
    // 扁平化
    val flatMapDS: DStream[String] = socketDS.flatMap(_.split(" "))

    // 结构转换 进行计数
    val mapDS: DStream[(String, Int)] = flatMapDS.map((_, 1))

    // 聚合操作
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)
    // 打印输出
    reduceDS.print()
    */

    resDS.print()

    // 启动采集器
    ssc.start()

    // 等待采集结束之后，终止程序
    ssc.awaitTermination()

  }
}
